package com.atguigu0.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @description: 窗口期输出操作
 * @time: 2020/6/15 20:54
 * @author: baojinlong
 **/
object WordCountStreaming6 {

  def main(args: Array[String]): Unit = {
    // 创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount")
    // 创建SteamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    // 必须设置ck
    ssc.checkpoint("E:/test-data/input/ck3333")
    // 创建Dstream
    val line: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    // 压平数据
    val word: DStream[String] = line.flatMap(_.split(" "))
    // 将单词转成元组
    val wordAndOne: DStream[(String, Int)] = word.map((_, 1))
    // 聚合四个批次的数据,每隔3s一次计算
  /*  val wordAndCount: DStream[(String, Int)] = wordAndOne.reduceByKeyAndWindow((x, y) => x + y, (x, y) => x - y,
      Seconds(12), Seconds(3), 4, (x: String) => x._2 > 0)
    // 打印数据
    wordAndCount.print
    // 开启sparkStreaming
    ssc.start()
    // 等待执行
    ssc.awaitTermination()*/
  }
}
